package start

import (
	"github.com/robfig/cron/v3"
	"github.com/spf13/viper"
	"log"
	"net/http"
	admin2 "src/api/subscribe/admin"
	"src/config"
	connection2 "src/connection"
	global2 "src/global"
	"src/task"
	"time"
)

func startHttp() {
	if config.V().GetBool("server.http.isOpen") {
		log.Println("启动 HTTP 服务 port:" + config.V().GetString("server.http.port"))
		//启动http服务
		router := HttpRouter()
		go log.Fatal(http.ListenAndServe(":"+config.V().GetString("server.http.port"), router))
	}
}
func startWebSocket() {

	startHeartBeatWebSocket()
	log.Println("启动 WEBSOCKET 服务 port:" + config.V().GetString("server.websocket.port"))
	router := WebSocketRouter()
	log.Println(http.ListenAndServe(":"+config.V().GetString("server.websocket.port"), router))

}
func startHeartBeatWebSocket() {
	//心跳多久没回应 视为掉线
	breakTimeout := viper.GetInt64("heartbeat.break-timeout")
	//格式化类型
	msgFmt := config.V().GetString("client.msg-fmt")
	//心跳间隔时间
	pingTime := viper.GetInt64("heartbeat.ping-time")
	pingTimeSecond := time.Second * time.Duration(pingTime)
	//房间在线时长记录
	c := cron.New()
	purification := &task.Purification{}
	c.AddFunc("@every "+pingTimeSecond.String(), func() {
		purification.HeartBeatFangWs(pingTime)
	})
	c.Start()
	//在线心跳
	d := cron.New()
	d.AddFunc("@every "+pingTimeSecond.String(), func() {
		purification.HeartBeatWs(breakTimeout, msgFmt)
	})
	d.Start()
}
func startTcp() {
	log.Println("启动 TCP 服务 port:" + config.V().GetString("server.tcp.port"))
	startHeartBeatTcp()
	router := TcpRouter()
	router.TcpHandler(config.V().GetString("server.tcp.ip"), config.V().GetInt("server.tcp.port"))

}
func startHeartBeatTcp() {
	//心跳多久没回应 视为掉线
	breakTimeout := viper.GetInt64("heartbeat.break-timeout")
	//格式化类型
	msgFmt := config.V().GetString("client.msg-fmt")
	//心跳间隔时间
	pingTime := viper.GetInt64("heartbeat.ping-time")
	pingTimeSecond := time.Second * time.Duration(pingTime)
	//房间在线时长记录
	e := cron.New()
	purification := &task.Purification{}
	e.AddFunc("@every "+pingTimeSecond.String(), func() {
		purification.HeartBeatFangTcp(pingTime)
	})
	e.Start()
	//在线心跳
	f := cron.New()
	f.AddFunc("@every "+pingTimeSecond.String(), func() {
		purification.HeartBeatTcp(breakTimeout, msgFmt)
	})
	f.Start()
}

// 启动kafka 订阅推送
func startSubKafka() {
	topicName := config.V().GetString("subscribe.kafka.topicName")
	log.Println("启动 KAFKA 订阅 服务 topicName::" + config.V().GetString("subscribe.Kafka.conf.topicName"))
	err := (&connection2.PubKafkaConsumer{}).NewKafkaConsumer().NewKafka()
	if err != nil {
		//return
	}
	////开启监听kafka 消息订阅
	(&connection2.PubKafkaConsumer{}).NewKafkaConsumer().Subscribe(topicName, admin2.NewKafka().SubChat)
}

// 启动redis 订阅推送
func startSubRedis() {
	channelName := config.V().GetString("subscribe.redis.channelName")
	log.Println("启动 REDIS 订阅 服务 channelName:" + channelName + ";ip:" + config.V().GetString("subscribe.redis.conf.ip") + ";pswd:" + config.V().GetString("subscribe.redis.conf.password"))

	////开启监听redis 消息订阅
	log.Println("Redis PubSubConn channelName:", channelName)
	//demo.RedisConsumer("chat")
	(&connection2.PubRedis{}).PubSubConn(channelName, admin2.NewRedis().SubChat)
}
func startFailInsertDb() {
	repeatRuntime := config.V().GetInt64("schedule-task.repeat-run-time")
	if repeatRuntime > 0 {
		purification := &task.Purification{}
		repeatRuntimeSecond := time.Second * time.Duration(repeatRuntime)
		log.Println("启动 心跳更新在在线时长，写入数据库失败重写任务，检测间隔：", repeatRuntime, "秒")
		//定时任务 更新失败的数据 重新执行
		g := cron.New()
		g.AddFunc("@every "+repeatRuntimeSecond.String(), func() {
			purification.RepeatRun()
		})
		g.Start()
	}
}

func Start() {
	//加载配置
	config.Init()
	//启动websocket 服务
	if config.V().GetBool("gateway.isOpen") {
		go StartGateway()
	}
	//全局变量
	global2.WsInit()
	global2.TcpInit()
	//connection.InIt() //初始化连接redis
	//日志格式
	log.SetFlags(log.Ldate | log.Ltime | log.Llongfile)

	go startHttp()
	purification := &task.Purification{}
	//启动websocket 服务
	if config.V().GetBool("server.websocket.isOpen") {
		go startWebSocket()
	}
	//启动tcp 服务
	if config.V().GetBool("server.tcp.isOpen") {
		go startTcp()
	}

	//启动 subscribe 订阅服务
	if config.V().GetBool("subscribe.redis.isOpen") {
		go startSubRedis()
	}
	//启动kafka 订阅服务
	if viper.GetBool("subscribe.Kafka.isOpen") {
		go startSubKafka()
	}
	startFailInsertDb()
	//服务关闭逻辑处理 kill 进程的时候拦截 结束所有任务后关闭
	purification.Serverdown()
	//pingTime := viper.GetInt64("heartbeat.ping-time")
	//pingTimeSecond := time.Second * time.Duration(pingTime)
	//log.Println("心跳时间间隔：" + pingTimeSecond.String())
	//c := cron.New()
	//c.AddFunc("@every "+pingTimeSecond.String(), (&utils.Purification{}).HeartBeat)
	//c.Start()
	//(&utils.Purification{}).HeartBeat()
	////开启监听kafka 消费组订阅 暂时不需要
	//go connection.KafkaConsumerGroup()

	//start.Start()
}
